package com.atguigu.gamll.realtime.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.java.tuple.Tuple2;
import redis.clients.jedis.Jedis;

import java.util.List;

// 查询维度工具类
public class DimUtil {

    public static JSONObject getDimInfo(String tableName, String id) {
        return getDimInfo(tableName, Tuple2.of("ID", id));
    }

    /**
     * 加上旁路缓存来获取维度数据
     * 旁路缓存：先从redis中获取数据，如果获取到了，这就是需要的维度数据，直接返回即可。
     *         如果没有获取到，那就要去Phoenix中获取数据，然后将获取到的数据缓存到redis中
     * Redis
     *      类型：String
     *      TTL：1天
     *      key：dim_表名:主键值1_主键值2
     */
    public static JSONObject getDimInfo(String tableName, Tuple2<String, String>... columnNameAndValues) {
        // 拼接要从redis中查询的key
        StringBuilder redisKey = new StringBuilder("dim:" + tableName.toLowerCase() + ":");
        // 拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "' ");
            redisKey.append(columnValue);
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
                redisKey.append("_");
            }
        }

        Jedis jedis = null;
        String dimStr = null;
        JSONObject dimJsonObj = null;

        try {
            jedis = RedisUtil.getJedis();
            dimStr = jedis.get(redisKey.toString());
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("从redis中查询失败");
        }

        // 判断是否从redis中获取到了维度数据
        if (dimStr != null && dimStr.length() > 0) {
            // 查询到了
            dimJsonObj = JSON.parseObject(dimStr);
        } else {
            // 没有查到，发送请求从Phoenix中查询
            System.out.println("查询的SQL语句是:" + selectSql);
            List<JSONObject> jsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);

            if (jsonObjList != null && jsonObjList.size() > 0) {
                dimJsonObj = jsonObjList.get(0);
                // 将查询到的数据缓存到redis中，方便下次查询
                if (jedis != null) {
                    jedis.setex(redisKey.toString(), 3600 * 24, dimJsonObj.toJSONString());
                }
            } else {
                System.out.println("维度表中没有找到相对应的数据");
            }
        }

        // 关闭连接
        if (jedis != null) {
            System.out.println("关闭Jedis");
            jedis.close();
        }
        return dimJsonObj;
    }

    public static JSONObject getDimInfoNoCache(String tableName, Tuple2<String, String>... columnNameAndValues) {
        // 拼接查询语句
        StringBuilder selectSql = new StringBuilder("select * from " + tableName + " where ");
        for (int i = 0; i < columnNameAndValues.length; i++) {
            Tuple2<String, String> columnNameAndValue = columnNameAndValues[i];
            String columnName = columnNameAndValue.f0;
            String columnValue = columnNameAndValue.f1;
            selectSql.append(columnName + "='" + columnValue + "' ");
            if (i < columnNameAndValues.length - 1) {
                selectSql.append(" and ");
            }
        }
        System.out.println("查询的SQL语句是:" + selectSql);

        List<JSONObject> jsonObjList = PhoenixUtil.queryList(selectSql.toString(), JSONObject.class);
        JSONObject jsonObj = null;
        if (jsonObjList != null && jsonObjList.size() > 0) {
            jsonObj = jsonObjList.get(0);
        } else {
            System.out.println("维度表中没有找到相对应的数据");
        }
        return jsonObj;
    }

    public static void main(String[] args) {
//        JSONObject dimInfo = getDimInfoNoCache("dim_base_trademark", Tuple2.of("id", "12"), Tuple2.of("tm_name", "chails"));
//        JSONObject dimInfo = getDimInfoNoCache("dim_base_trademark", Tuple2.of("id", "12"));
        JSONObject dimInfo = getDimInfo("dim_base_trademark", "15");
        System.out.println(dimInfo);
    }

    public static void delCached(String tableName, String key) {
        String redisKey = "dim:" + tableName.toLowerCase() + ":"+key;
        try {
            Jedis jedis = RedisUtil.getJedis();
            jedis.del(redisKey);
            jedis.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
